package com.shujia.sink

import java.sql.{Connection, DriverManager}

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.api.scala._

object Demo3SinkMysql {
  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val ds = env.socketTextStream("node1", 8888)

    val countDS = ds
      .flatMap(_.split(","))
      .map(word => (word, 1))
      .keyBy(_._1)
      .reduce((x, y) => (x._1, x._2 + y._2))

    //将数据sink到mysql
    countDS.addSink(new MysqlSink)

    env.execute("Demo3SinkMysql")

  }

}

class MysqlSink extends RichSinkFunction[(String, Int)] {

  var connector: Connection = _

  override def open(parameters: Configuration): Unit = {
    //打开连接
    Class.forName("com.mysql.jdbc.Driver")
    connector = DriverManager.getConnection("jdbc:mysql://node1:3306/student", "root", "123456")

  }

  override def close(): Unit = {
    if (connector != null) {
      connector.close()
    }
  }

  override def invoke(kv: (String, Int), context: SinkFunction.Context[_]): Unit = {

    val sql = "insert into t_word(word,count) values(?,?)"

    val stat = connector.prepareStatement(sql)

    stat.setString(1, kv._1)
    stat.setInt(2, kv._2)

    stat.executeUpdate()

  }
}
